package com.we.flink.checkpoint;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import com.we.flink.bean.FlinkCheckpointInfo;
import com.we.flink.function.FlinkCheckpointKeyedProcessFunction;
import com.we.flink.function.MapFlinkRestAPIFunction;
import com.we.flink.source.YarnApplicationMonitorSource;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.SimpleStringSerializer;

import java.util.Properties;

/**
 * @Author: JasonLee @Description: 通过读取 Yarn 上运行任务调用 Flink Rest API 获取 任务对应的 checkpoint path 信息,保存到
 * Doris 里面
 */
public class GetFlinkCheckpointPath {
    private static final String JOB_NAME = GetFlinkCheckpointPath.class.getSimpleName();
    private static final String PROD_FE_NODES =
            "10.10.148.68:8030,10.10.148.69:8030,10.10.148.74:8030";
    private static final String PROD_DORIS_USERNAME = "root";
    private static final String PROD_TABLE_IDENTIFIER = "flink_job_info.flink_job_checkpoint_info";
    private static final String PROD_DORIS_PASSWORD = "";
    /** 测试环境的配置 */
    private static final String TEST_FE_NODES = "172.16.1.91:18030";

    private static final String TEST_DORIS_USERNAME = "root";
    private static final String TEST_DORIS_PASSWORD = "root!123";
    private static final String TEST_TABLE_IDENTIFIER = "test_db.flink_job_checkpoint_info";

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        final SingleOutputStreamOperator<String> map =
                env.addSource(new YarnApplicationMonitorSource(), "Yarn Application Monitor Source")
                        .map(new MapFlinkRestAPIFunction())
                        .keyBy(FlinkCheckpointInfo::getApplicationId)
                        .process(new FlinkCheckpointKeyedProcessFunction());

        DorisSink.Builder<String> builder = DorisSink.builder();
        DorisOptions.Builder dorisBuilder = DorisOptions.builder();
        dorisBuilder
                .setFenodes(PROD_FE_NODES)
                .setTableIdentifier(PROD_TABLE_IDENTIFIER)
                .setUsername(PROD_DORIS_USERNAME)
                .setPassword(PROD_DORIS_PASSWORD);

        Properties properties = new Properties();
        properties.setProperty("format", "json");
        properties.setProperty("read_json_by_line", "true");

        DorisExecutionOptions.Builder executionBuilder = DorisExecutionOptions.builder();
        // streamload label prefix
        executionBuilder
                .setLabelPrefix("label-doris_" + System.currentTimeMillis())
                .setStreamLoadProp(properties);

        final DorisSink<String> build =
                builder.setDorisReadOptions(DorisReadOptions.builder().build())
                        .setDorisExecutionOptions(executionBuilder.build())
                        // serialize according to string
                        .setSerializer(new SimpleStringSerializer())
                        .setDorisOptions(dorisBuilder.build())
                        .build();

        map.sinkTo(build).name("Doris Sink");
        env.execute(JOB_NAME);
    }
}
